package com.yl.flink.processor;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.yl.constant.Const;
import com.yl.constant.TimeUnit;
import com.yl.entity.MultiDataEntity;
import com.yl.util.DtUtil;
import com.yl.util.FlinkUtil;
import com.yl.util.InfluxUtil;
import com.yl.util.SUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBConfig;
import org.apache.flink.util.Collector;
import org.influxdb.InfluxDB;
import org.influxdb.dto.QueryResult;

import java.util.ArrayList;
import java.util.List;

public class HourCountFunc extends KeyedProcessFunction<String, MultiDataEntity, Tuple4<String, String, Long, Integer>> {

    // 缓存需要重新统计的时间范围，开始时间和结束时间（精确到小时）
    private transient ValueState<String> startTime;
    private transient ValueState<String> endTime;
    private transient InfluxDB influxDBClient;
    private final InfluxDBConfig influxDBConfig;

    public HourCountFunc(ParameterTool params) {
        this.influxDBConfig = FlinkUtil.getInfluxDBConfig(params);
    }

    /**
     * 初始化资源配置
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        startTime = getRuntimeContext().getState(new ValueStateDescriptor<>(Const.S_COUNT_CAL_ST, TypeInformation.of(new TypeHint<>() {
        })));
        endTime = getRuntimeContext().getState(new ValueStateDescriptor<>(Const.S_COUNT_CAL_ET, TypeInformation.of(new TypeHint<>() {
        })));
        this.influxDBClient = InfluxUtil.connect(this.influxDBConfig);
    }

    /**
     * 每隔10分钟把缓存的时间段的数据条数重新统计一次
     */
    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, MultiDataEntity, Tuple4<String, String, Long, Integer>>.OnTimerContext ctx, Collector<Tuple4<String, String, Long, Integer>> out) throws Exception {
        // 查询数据库获取缓存时间范围内该测点的小时数据量统计
        reCalHourCount(ctx.getCurrentKey(), startTime.value(), endTime.value())
                .forEach(out::collect);
        // 执行完任务删除定时器
        ctx.timerService().deleteProcessingTimeTimer(timestamp);
        // 然后把此次计算过的缓存信息清除掉
        startTime.clear();
        endTime.clear();
    }

    /**
     * 从历史数据中获取需要重新统计的时间范围
     * 输出结果Tuple3<微秒级小时时间戳，测点id，小时数据条数>
     */
    @Override
    public void processElement(MultiDataEntity element, KeyedProcessFunction<String, MultiDataEntity, Tuple4<String, String, Long, Integer>>.Context ctx, Collector<Tuple4<String, String, Long, Integer>> out) throws Exception {
        // 精确到小时
        String pickTime = DtUtil.toDateStart(element.getPickTime(), TimeUnit.H);
        if (null != pickTime) {
            // 向后偏移一小时，这样才能覆盖涉及到的所有数据
            String et = DateUtil.offsetHour(DateUtil.parseDate(pickTime), 1).toString();
            // 首先判断是否已经缓存了时间范围
            if (null != startTime.value()) {
                // 如果已经缓存过了，判断缓存的结束时间是否早于此次数据时间向后加一小时，是就用此次时间向后加一小时作为结束时间
                if (DateUtil.parse(endTime.value()).getTime() < DateUtil.parse(et).getTime()) {
                    endTime.update(et);
                }
            } else {
                // 如果还未缓存，把此次数据时间作为开始时间，向后加一小时作为结束时间
                startTime.update(pickTime);
                endTime.update(et);
                // 注册定时器，默认10分钟后触发
                // TODO: 2022/10/8 测试调整为一分钟
                long fireTime = ctx.timerService().currentProcessingTime() + 60_000;
//                long fireTime = ctx.timerService().currentProcessingTime() + 600_000;
                ctx.timerService().registerProcessingTimeTimer(fireTime);
            }
        }
    }

    /**
     * 获取该测点指定时间范围内每个小时的条数统计
     *
     * @param key       分组key
     * @param startTime 开始时间
     * @param endTime   结束时间
     */
    private Iterable<Tuple4<String, String, Long, Integer>> reCalHourCount(String key, String startTime, String endTime) {
        List<Tuple4<String, String, Long, Integer>> res = new ArrayList<>();
        String typeTag = SUtil.deFmt(key, 1);
        String targetId = SUtil.deFmt(key, 2);
        String sql = StrUtil.format(Const.INFLUXDB_SQL_COUNT, targetId, startTime, endTime);
        List<QueryResult.Series> seriesList = InfluxUtil.executeSql(this.influxDBClient, this.influxDBConfig.getDatabase(), sql);
        for (QueryResult.Series series : seriesList) {
            for (List<Object> value : series.getValues()) {
                long microTs = DtUtil.microTs(value.get(0).toString());
                int hourCount = Integer.parseInt(value.get(2).toString());
                res.add(Tuple4.of(typeTag, targetId, microTs, hourCount));
            }
        }
        return res;
    }

    @Override
    public void close() {
        if (this.influxDBClient.isBatchEnabled()) {
            this.influxDBClient.disableBatch();
        }
        this.influxDBClient.close();
    }

}
